package ranks.tasks.count_seeds;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashSet;
import java.util.Set;

import mapred.example.base.TextMapper;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import ranks.base.Util;
import external.Host;

public class CountSeeds_Mapper extends TextMapper {

	private static Set<String> seeds = new HashSet<String>();
	
	@Override
	public void map(Text key, Text value, OutputCollector<Text, Text> output,
			Reporter reporter) throws IOException {
		reporter.incrCounter("WebSize", "Number", 1);
		if (seeds.contains(Host.getHost(key.toString()))) {
			reporter.incrCounter("Seeds", "Number", 1);
		}
	}

	@Override
	public void setUp() {
		try {
			FileSystem fs = FileSystem.get(this.getJobConf());
			BufferedReader br = new BufferedReader(new InputStreamReader(fs
					.open(new Path(this.getJobConfConfig("Seeds")))));
			for (String line = br.readLine(); line != null; line = br.readLine()) {
				seeds.add(Util.getHost(line.trim()));
			}
			br.close();
			seeds.remove("");
		} catch (IOException e) {
			e.printStackTrace();
			System.exit(255);
		}
	}

}
